package server

import (
	"encoding/binary"
	"fmt"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"gse/transport"
	gseErr "gse/transport/errors"
	"reflect"
	"sync"
	"time"
)

// DefaultWorkerIdleTime in milliseconds
const DefaultWorkerIdleTime = 10000

var _logger *zap.Logger
var loggerChangeLock = new(sync.RWMutex)

type GSEServer interface {
	Serve()
	RegisterServiceHandler(serviceName string, handler ServiceHandler, paramType proto.Message)
	RegisterEvent(name string, cfg *EventConfig)
	GetEventPusher() *EventPusher
}
type ServerBase struct {
	serviceHandlerMap *sync.Map
	transport         transport.Transport
	sessions          *sync.Map
	sesCfg            *SessionConfig
	workLoadChan      chan *workLoad
	pusher            *EventPusher
}

type workLoad struct {
	ses *Session
	req *Request
}

func NewServerBase(t transport.Transport, sesCfg *SessionConfig) *ServerBase {
	return &ServerBase{
		transport:    t,
		workLoadChan: make(chan *workLoad),
		sesCfg:       sesCfg,
	}
}

func init() {
	_logger, _ = zap.NewDevelopment()
}

func logger() *zap.Logger {
	loggerChangeLock.RLock()
	l := _logger
	loggerChangeLock.RUnlock()
	return l
}

func SetLogger(logger *zap.Logger) {
	loggerChangeLock.Lock()
	_logger = logger
	loggerChangeLock.Unlock()
}

func (s *ServerBase) RegisterEvent(name string, cfg *EventConfig) {
	s.pusher.RegisterEvent(name, cfg)
}

func (s *ServerBase) RegisterServiceHandler(serviceName string, handler ServiceHandler, paramType proto.Message) {
	info := &ServiceHandlerRegisterInfo{
		Handler:   handler,
		paramType: reflect.TypeOf(reflect.ValueOf(paramType).Elem().Interface()),
	}
	s.serviceHandlerMap.Store(serviceName, info)
	logger().Debug("registered rpc handler", zap.String("name", serviceName), zap.String("param", fmt.Sprintf("%v", info.paramType)))
}

func GetLogger() *zap.Logger {
	return logger()
}

func (s *ServerBase) Init() {
	s.serviceHandlerMap = new(sync.Map)
	s.sessions = new(sync.Map)
	s.pusher = NewEventPusher(
		&EventPusherConfig{},
	)
	s.sesCfg.RegisterHook(AfterClose(func(ses *Session, args ...interface{}) {
		s.sessions.Delete(ses.conn.RemoteAddr().String())
		logger().Debug("session closed, removing: " + ses.conn.RemoteAddr().String())
	}))
	s.sesCfg.RegisterHook(OnRequest(func(req *Request, param proto.Message, ses *Session, args ...interface{}) (rsp proto.Message, err error) {
		switch req.serviceName {
		case "#unsub":
			rsp = s.pusher.OnUnSub(param)
		case "#sub":
			rsp = s.pusher.OnNewSub(ses, param)
		}
		return
	}))
}

func (s *ServerBase) Serve() {
	for {
		conn, err := s.transport.Accept()
		if err != nil {
			panic(err)
		}
		go s.handleConn(conn)
	}
}

func (s *ServerBase) GetEventPusher() *EventPusher {
	return s.pusher
}

func (s *ServerBase) handleConn(conn transport.Connection) {
	sesI, ok := s.sessions.Load(conn.RemoteAddr().String())
	var ses *Session
	if !ok {
		sesCfg := *s.sesCfg
		ses = NewSession(conn, &sesCfg, &s.serviceHandlerMap)
		s.sessions.Store(conn.RemoteAddr().String(), ses)
		logger().Debug("created new session", zap.String("remote_addr", conn.RemoteAddr().String()))
	} else {
		ses = sesI.(*Session)
	}

	for {
		rpcObject, err := s.recvRpc(conn)
		if err != nil {
			if err == gseErr.ConectionClosed {
				logger().Debug(fmt.Sprintf("conn closed: %v, %v", conn.RemoteAddr(), err))
				ses.Close()
				return
			}
			logger().Warn(err.Error())
			continue
		}

		switch t := rpcObject.(type) {
		case *Request:
			s.DispatchRequest(ses, t)
		case *Response:
			logger().Debug("Got request response:", zap.Uint16("trans_id", t.GetTransactionID()), zap.String("remote_addr", conn.RemoteAddr().String()))
			respWaithChanI, ok := ses.rspContainer.Load(t.transactionID)
			if !ok {
				logger().Warn("No transaction found for response, data dropped.",
					zap.Uint16("trans_id", t.GetTransactionID()),
					zap.String("remote_addr", conn.RemoteAddr().String()),
				)
				continue
			}
			respWaithChanI.(chan *Response) <- t
		case *Push:
			ses.ReceiveEventPush(t)
		default:
			logger().Warn(fmt.Sprintf("unknown remote data type: %+v", t))
		}
	}
}

func (s *ServerBase) recvRpc(conn transport.Connection) (RpcObj, error) {
	data, err := conn.ReadData()
	if err != nil {
		return nil, err
	}
	if len(data) < 7 {
		return nil, fmt.Errorf("received broken packet: data length should not be less than 7, but got %v", len(data))
	}
	offset := 0
	protoVer := data[offset] >> 3
	actionType := LogicPacketType(data[offset] & 7)
	if protoVer != ProtoVerNo {
		return nil, fmt.Errorf("received broken packet: ver number err, expects %v, but got %v.", ProtoVerNo, data[offset])
	}

	offset += 1
	dataLength := binary.BigEndian.Uint32(data[offset : offset+4])
	offset += 4
	if uint32(len(data))-5 != dataLength {
		return nil, fmt.Errorf("received broken packet: data length not match, expects: %v, but got %v", dataLength, len(data)-5)
	}

	if actionType == LogicPacketEventPush {
		rp := &Push{}
		rp.InitData(data[offset:])
		return rp, nil
	}

	transactionID := binary.BigEndian.Uint16(data[offset : offset+2])
	offset += 2
	isResponse := transactionID%2 == 0
	if isResponse {
		return NewResponse(transactionID, RpcResponseStatus(data[offset]), data[offset+1:]), nil
	}
	serviceNameLength := int(data[offset])
	offset += 1
	return NewRequest(string(data[offset:offset+serviceNameLength]), transactionID, data[offset+serviceNameLength:]), nil
}

// HandleRequest invoke the registered service handler to process the request and send the response to the remote caller.
// TODO: grace exit.
func (s *ServerBase) DispatchRequest(ses *Session, req *Request) {
	wkl := &workLoad{
		req: req,
		ses: ses,
	}

	if le := logger().Check(zap.DebugLevel, "dispatching workload"); le != nil {
		le.Write(zap.String("remote_addr",
			ses.conn.RemoteAddr().String()),
			zap.String("request_name", req.GetServiceName()),
			zap.Uint16("tid", req.GetTransactionID()),
		)
	}

	timeout := time.Millisecond * DefaultWorkerIdleTime

	select {
	// try to dispatch req to worker.
	case s.workLoadChan <- wkl:
		return
	// if no workers are available, then create one.
	default:
		go func(wkl *workLoad) {
			t := time.NewTimer(timeout)
		PROCESS:
			{
				err := wkl.ses.HandleRequest(wkl.req)
				if err != nil {
					logger().Warn("Failed to handle request.", zap.Error(err),
						zap.String("service_name", wkl.req.GetServiceName()),
						zap.String("remote_addr", wkl.ses.conn.RemoteAddr().String()),
						zap.Uint16("tid", wkl.req.GetTransactionID()),
					)
				}
				t.Stop()
				t.Reset(timeout)
			}

			for {
				select {
				case <-t.C:
					logger().Debug("worker idle timedout, exit...", zap.Duration("elapsed", timeout))
					return
				case wkl = <-s.workLoadChan:
					goto PROCESS
				}
			}
		}(wkl)
	}
}
